package com.cupdata.marathon.common;

import cn.hutool.core.util.StrUtil;
import com.cupdata.marathon.client.AcctQueryService;
import com.cupdata.marathon.definition.OperQueryRslt;
import com.cupdata.marathon.definition.RspCd;
import com.cupdata.marathon.entity.TblHackCouponDownTransLog;
import com.cupdata.marathon.entity.TblRetryTransLog;
import com.cupdata.marathon.param.acctquery.AcctQueryRequest;
import com.cupdata.marathon.param.acctquery.AcctQueryResponse;
import com.cupdata.marathon.service.ITblHackCouponDownTransLogService;
import com.cupdata.marathon.utils.GsonUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;

@Component
public class KafakaBatchReceiver {

    @Autowired
    private ITblHackCouponDownTransLogService hackCouponDownTransLogService;

    @Autowired
    private AcctQueryService acctQueryService;

    @Autowired
    private ITblHackCouponDownTransLogService downTransLogService;

    @Autowired
    private RedisService redisService;

    @KafkaListener(id = "kafka_batch_trans", topics = "${topicName.topicTrans}")
    public void receiveTrans(List<ConsumerRecord<String, String>> records) {
        hackCouponDownTransLogService.saveOrUpdateBatch(records.stream()
                .map(r -> GsonUtils.gson.fromJson(r.value(), TblHackCouponDownTransLog.class))
                .collect(Collectors.toList()));
    }

    @KafkaListener(id = "kafka_batch_retry", topics = "${topicName.topicRetry}")
    public void receiveRetry(List<ConsumerRecord<String, String>> records) {
        records.forEach(r -> {

            TblRetryTransLog log = GsonUtils.gson.fromJson(r.value(), TblRetryTransLog.class);
            AcctQueryRequest acctQueryRequest = new AcctQueryRequest();
            acctQueryRequest.setAcctTp(log.getAcctTp());
            acctQueryRequest.setTransId(log.getTransId());
            acctQueryRequest.setUserId(log.getUserId());
            acctQueryRequest.setCouponId(log.getCouponId());
            AcctQueryResponse acctQueryResponse = acctQueryService.call(acctQueryRequest);
            if (RspCd.SUCCESS.getCode().equals(acctQueryResponse.getRspCd())) {
                if (StrUtil.equalsAny(acctQueryResponse.getOperRslt(),
                        OperQueryRslt.SUCCESS.getCode(),
                        OperQueryRslt.FAIL.getCode(),
                        OperQueryRslt.ILLEGAL.getCode())) {
                    execute(log, acctQueryResponse);
                }
            }
        });
    }

    private void execute(TblRetryTransLog record, AcctQueryResponse response) {
        // 更新领取流水表
        downTransLogService.update(record.getCouponId(), response);

        if (!OperQueryRslt.SUCCESS.getCode().equals(response.getOperRslt())) {
            // todo 操作是否可以合并
            // 增加优惠券总数
            // 删除该用户领取的券
            redisService.recoverCoupon(record.getUserId(), record.getCouponTp());
        }
    }
}
